package com.jumei.flume.serializer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.Configurable;
import org.apache.flume.serialization.EventSerializer;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.Charset;

/**
 * Created by minj on 2017/8/28.<br/>
 */
public class Json2SplitEventSerializer implements EventSerializer, Configurable {

    private static final Logger logger = LoggerFactory.getLogger(Json2SplitEventSerializer.class);

    private final String COLUMNS_NAME = "columns";

    private final String SPILT_NAME = "spilt";
    private final static String SPILT_DFLT = "|";
    private final String APPEND_WRITE_TIME = "appendWriteTime";
    private final boolean APPEND_WRITE_TIME_DFLT = true;
    // for legacy reasons, by default, append a newline to each event written out
    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean APPEND_NEWLINE_DFLT = true;

    private final OutputStream out;
    private String [] columns;//对应字段
    private String spilt;//分割字符
    private boolean appendNewline;
    private boolean appendWriteTime;//是否增加写入时间放在第一位

    public static Builder builder() {
        return new Builder();
    }

    public Json2SplitEventSerializer(OutputStream out, Context context) {
        this.out = out;
        this.configure(context);
    }

    @Override
    public void configure(Context context) {
        String columnsStr = context.getString(COLUMNS_NAME);
        if (columnsStr == null || columnsStr.isEmpty()) {
            throw new IllegalArgumentException("Json2Split columns empty");
        }

        this.columns = columnsStr.split(",");
        this.appendNewline = context.getBoolean(APPEND_NEWLINE, APPEND_NEWLINE_DFLT);
        this.spilt = context.getString(SPILT_NAME, SPILT_DFLT);
        this.appendWriteTime = context.getBoolean(APPEND_WRITE_TIME, APPEND_WRITE_TIME_DFLT);

        logger.info("configure: columns {},appendNewline {}, spilt {}, appendWriteTime {}", new Object[]{columnsStr, spilt, appendNewline, appendWriteTime});
    }

    @Override
    public void afterCreate() throws IOException {
        // no-op
        logger.info("Json2SplitEventSerializer afterCreate");
    }

    @Override
    public void afterReopen() throws IOException {
        throw new UnsupportedOperationException("Json2Split API doesn't support append");
    }

    @Override
    public void write(Event event) throws IOException {

        byte[] body = event.getBody();
        String message = new String(body, Charset.defaultCharset());
        if (StringUtils.isBlank(message)) {
            logger.warn("empty body");
            return;
        }

        try {
            JSONObject json = JSON.parseObject(message);
            StringBuffer sb = new StringBuffer();
            if (appendWriteTime) {
                sb.append(DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
                sb.append(spilt);
            }

            for (String column : columns) {
                Object o = json.get(column.trim());
                if (o == null) {
                    sb.append("");
                } else {
                    sb.append(filterCharacter(o.toString()));
                }
                sb.append(spilt);
            }
            String s = sb.substring(0, sb.length() - spilt.length());
            if (appendNewline) {
                s += "\n";
            }
            out.write(s.getBytes(Charset.defaultCharset()));
        } catch (Exception e) {
            logger.error("Json2Split write error", e);
            throw  e;
        }

    }

    //过滤字符串隐藏的特殊字符
    private String filterCharacter(String s) {
        if (s == null) {
            return null;
        }

        return s.replaceAll("\n|\r|\t", "");
    }

    @Override
    public void flush() throws IOException {
        logger.debug("Json2SplitEventSerializer flush");
    }

    @Override
    public void beforeClose() throws IOException {
        logger.info("Json2SplitEventSerializer beforeClose");
    }

    @Override
    public boolean supportsReopen() {
        logger.info("Json2SplitEventSerializer supportsReopen");
        return true;
    }

    public static class Builder implements EventSerializer.Builder {

        @Override
        public EventSerializer build(Context context, OutputStream out) {
            logger.info("Json2SplitEventSerializer build");
            return new Json2SplitEventSerializer(out, context);
        }

    }

}
